package com.rem.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 发送数据到kafka
 * 分区
 *
 * @author Rem
 * @date 2022-12-28
 */
public class CustomerProducerPatitioner {

    public static void main(String[] args) {

        //specifyKey();
        //specifyPartition();
        specifyPatitioner();

    }

    /**
     * 指定发送消息的key
     */
    private static void specifyKey() {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");

        //配置kafka连接器以及key value序列化方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties, new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 5; i++) {
            /**
             * 创建要发送到指定主题和分区的记录
             * topic -记录追加到的topic
             * key -记录中包含的key
             * value -记录内容
             */
            String key;
            if (i % 3 == 0) {
                key = "a";
            } else if (i % 3 == 1) {
                key = "b";
            } else {
                key = "f";
            }
            kafkaProducer.send(new ProducerRecord<>("first", key, "specifyKey" + i));
        }
        kafkaProducer.close();
    }


    /**
     * 指定分区
     */
    private static void specifyPartition() {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");

        //配置kafka连接器以及key value序列化方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties, new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 5; i++) {
            /**
             * 创建要发送到指定主题和分区的记录
             * topic -记录追加到的topic
             * partition -记录应该发送到的分区
             * key -记录中包含的key
             * value -记录内容
             */
            kafkaProducer.send(new ProducerRecord<>("first", 1, "", "specifyPartition" + i));
        }
        kafkaProducer.close();
    }

    /**
     * 自定义分区器
     */
    private static void specifyPatitioner() {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");

        // 关联自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getCanonicalName());

        //配置kafka连接器以及key value序列化方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties, new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 5; i++) {
            String value = "specifyPatitioner";
            if (i % 2 == 0) {
                value = "rem";
            }
            kafkaProducer.send(new ProducerRecord<>("first", value + i));
        }
        kafkaProducer.close();
    }
}
